/****************************************************************************
 *   Copyright (C) 2008 by 7552                                             *
 *   fiubataller2-08c1@gruposyahoo.com.ar                                   *
 *                                                                          *
 *  Licensed under the Apache License, Version 2.0 (the "License");         *
 *  you may not use this file except in compliance with the License.        *
 *  You may obtain a copy of the License at                                 *
 *                                                                          *
 *      http://www.apache.org/licenses/LICENSE-2.0                          *
 *                                                                          *
 *  Unless required by applicable law or agreed to in writing, software     *
 *  distributed under the License is distributed on an "AS IS" BASIS,       *
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.*
 *  See the License for the specific language governing permissions and     *
 *  limitations under the License.                                          *
 *                                                                          *
 ***************************************************************************/

#include "TransactionManager.h"
#include "LogRecords.h"
#include "IScheduler.h"
#include "CriticalSection.h"


#include <stdlib.h>
#include <ctime>
#include <iostream> //para debug


using namespace Taller2::TransRec;

const size_t MAX_FILE_LOG_SIZE = 200 * 1024 * 1024; // 200 Mb

TransactionManager::TTransactionManagerMap TransactionManager::instances;

TransactionManager* TransactionManager::Instance(IScheduler* pScheduler, const string& database)
{
	pthread_mutex_t mutex =
	PTHREAD_MUTEX_INITIALIZER;
	CCriticalSection cs(mutex, true);

	if (instances.empty())
	{
		TransactionManager* lInstance = new TransactionManager(database);
		instances[database] = lInstance;
		lInstance->scheduler = pScheduler;
		return lInstance;
	}
	else
	{
		for (TTransactionManagerMap::iterator it = instances.begin(); it != instances.end(); it++)
		{
			if (it->first == database)
				return it->second;
		}
		TransactionManager* lInstance = new TransactionManager(database);
		instances[database] = lInstance;
		lInstance->scheduler = pScheduler;
		return lInstance;
	}
}

void TransactionManager::CloseTransactionManager(const string& database)
{
	//TODO: Ver si libera el puntero
	TransactionManager* lInstance = instances[database];
	instances.erase(database);
	delete lInstance;
}

//***************************************************************************
TransactionManager::TransactionManager(const string &database) :
	logCache(database)
{
	srand(time(NULL));
	nextTransactionId = rand();

	ContOpe=0;

	CCriticalSection::CreateCriticalSection(critical_section);

	//Todavia no esta listo...
	//if (logCache.GetFileSize() > MAX_FILE_LOG_SIZE)
	//	logCache.CleanUpLog();
}

//***************************************************************************

TransactionManager::~TransactionManager()
{
	CCriticalSection::DeleteCriticalSection(critical_section);
}

//***************************************************************************

int TransactionManager::BeginTransaction()
{

	CCriticalSection cs(critical_section);

	return nextTransactionId++;
}

//***************************************************************************

int TransactionManager::CommitTransaction(const int transactionId)
{
	CCriticalSection cs(critical_section);

	CCommitRecord* record = new CCommitRecord(transactionId, time(NULL));
	logCache.WriteEntry(record);
	logCache.Flush();
	return NO_ERROR;
}

//***************************************************************************

int TransactionManager::RollbackTransaction(const int transactionId)
{
	CCriticalSection cs(critical_section);

	CRollbackRecord* record = new CRollbackRecord(transactionId, time(NULL));
	logCache.WriteEntry(record);
	logCache.Flush();

	if (scheduler)
	{
		CBasicOperationRecord* rec;
		rec=NULL;
		logCache.FindRollback(transactionId, rec);
		try
		{

			while (logCache.GetPrevRecordOfSameTransaction(rec))
			{ // FIXME Pasar el ClassId al oid

				if (rec->EntryType() == INSERT)
				{
					CInsertRecord* lTemp = reinterpret_cast<CInsertRecord*> (rec);
					ObjectId oid;
					oid.instanceId = lTemp->ObjectID();
					scheduler->deleteObject(oid);
				}
				else if (rec->EntryType() == DELETE)
				{
					CDeleteRecord* lTemp = reinterpret_cast<CDeleteRecord*> (rec);
					ObjectId oid;
					oid.instanceId = lTemp->ObjectID();
					scheduler->createObject(lTemp->Object(), oid);
				}
				else if (rec->EntryType() == UPDATE)
				{
					CUpdateRecord* lTemp = reinterpret_cast<CUpdateRecord*> (rec);
					scheduler->updateObject(lTemp->OldObject());
				}
			}
			if (rec)
				delete rec;
			return NO_ERROR;
		} catch (...)
		{
			// ERROR durante el rollback..
			return ERR_ROLLBACK;
		};
	}
	else
	{
		//No se puede obtener una instancia con Operaciones
		return ERR_ROLLBACK;
	};
	return NO_ERROR;
}

//***************************************************************************

int TransactionManager::PreWrite(const long long objectId, const char* oldObject, const int oldObjectSize,
		const char* newObject, const int newObjectSize)
{
	CCriticalSection cs(critical_section);

	int transactionId = BeginTransaction();
	if (PreWrite(objectId, oldObject, oldObjectSize, newObject, newObjectSize, transactionId))
		return CommitTransaction(transactionId);
	else
		return ERR_PREWRITE;
}

//***************************************************************************

int TransactionManager::PreWrite(const long long objectId, const char* oldObject, const int oldObjectSize,
		const char* newObject, const int newObjectSize, const int transactionId)
{
	CCriticalSection cs(critical_section);

	if (oldObject == NULL && newObject == NULL)
		throw "oldObject y newObject no pueden ser ambos punteros nulos";

	CLogRecord* record = 0;

	if (oldObject == NULL)
	{
		record = new CInsertRecord(transactionId, time(NULL), objectId);
		CInsertRecord* insertRecord = reinterpret_cast<CInsertRecord*> (record);
		insertRecord->SetState(newObject, newObjectSize);
	}
	else if (newObject == NULL)
	{
		record = new CDeleteRecord(transactionId, time(NULL), objectId);
		CDeleteRecord* deleteRecord = reinterpret_cast<CDeleteRecord*> (record);
		deleteRecord->SetState(oldObject, oldObjectSize);
	}
	else
	{
		record = new CUpdateRecord(transactionId, time(NULL), objectId);
		CUpdateRecord* updateRecord = reinterpret_cast<CUpdateRecord*> (record);
		updateRecord->SetState(oldObject, oldObjectSize, newObject, newObjectSize);
	}

	if (record)
		{
			ContOpe++;

			//POLITICA DE GENERACION DE CHECKPOINT
			if(ContOpe>CHECKPOINT_INTERVAL)	
				{
					WriteCheckpoint();
					ContOpe=0;			
				}
			return logCache.WriteEntry(record);
		}
	else
		return ERR_PREWRITE;
}

//***************************************************************************

void TransactionManager::PreFlushCache(const list<long long>& objectIdsToFlush)
{
	logCache.Flush();
}

//***************************************************************************

void TransactionManager::WriteCheckpoint()
{
	CCriticalSection cs(critical_section);

	CCheckPointRecord *record = new CCheckPointRecord(time(NULL));
	logCache.GetOpenTransactionsInfo(record);
	logCache.WriteEntry(record);
	logCache.Flush();
}
//***************************************************************************
